package juc;

// 获取随机数
import static java.util.concurrent.ThreadLocalRandom.current;

import java.util.Date;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.IntStream;

/**
 * Semaphore（信号量）
 * 用途：允许限定多个线程对共享资源进行并行操作
 * 许可证数量（permits）：可以一次获取一个或多个，获取成功一次，减少对应数量的许可证，无法获取超过剩余总数的许可证。
 * @author ben
 * @date 2021-09-10 21:10:59 CST
 */
public class TestSemaphore {
	
	public static Consumer<Object> cs = System.out::println;

	public static void main(String[] args) {
//		test1();
		
//		test2();
		
//		test3();
		
		test4();
	}

	/**
	 * 试验1：基本使用
	 * @author ben
	 * @date 2021-09-10 21:29:52 CST
	 */
	public static void test1() {
		cs.accept("test1() start");
		SemaphoreTest1 test = new SemaphoreTest1();
		try {
			test.basicUse();
		} catch (InterruptedException e) {
			cs.accept("test.basicUse()异常：e=" + e.getMessage());
		}
		cs.accept("test1() end");
	}
	
	/**
	 * 试验2：控制使用共享资源的最大线程数
	 * @author ben
	 * @date 2021-09-10 22:06:11 CST
	 */
	public static void test2() {
		cs.accept("test2() start");
		final int tnums = 10;
		ExecutorService es = Executors.newFixedThreadPool(tnums);
		
		SemaphoreTest2 t2 = new SemaphoreTest2();
		
		IntStream.range(0, tnums).forEach(item->{
			es.execute(()->{
				// 每个线程不停滴尝试使用 useStar，但同时只有5个可以使用
				String tname = Thread.currentThread().getName();
				cs.accept(tname + ": GO!");
				while(true) {
					try {
						t2.useStar();
					} catch (InterruptedException e1) {
						cs.accept(tname + ": e1=" + e1.getMessage());
					}
					
					try {
						TimeUnit.SECONDS.sleep(1);
					} catch (InterruptedException e2) {
						cs.accept(tname + ": e2=" + e2.getMessage());
					}
				}
			});
		});
		cs.accept("test2() end");
	}
	
	/**
	 * 试验3：增强的Semaphore——只有获取了许可证的才可以释放许可
	 * 更进一步：只允许释放少于等于获取数量的许可
	 * @author ben
	 * @date 2021-09-10 22:51:15 CST
	 */
	public static void test3() {
		cs.accept("test3() start");
		
		// 本试验进行时，只重写了 acquire(), release() 两个函数
		SemaphoreTest3 t3 = new SemaphoreTest3(1);
		cs.accept("0 availablePermits=" + t3.availablePermits());
		// 无权释放
		t3.release();
		cs.accept("1 availablePermits=" + t3.availablePermits());
		
		try {
			t3.acquire();
			cs.accept("2 availablePermits=" + t3.availablePermits());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		t3.release();
		cs.accept("3 availablePermits=" + t3.availablePermits());


		// 仅1个许可证，下面执行两次获取，预期：程序阻塞
		try {
			t3.acquire();
			cs.accept("4 availablePermits=" + t3.availablePermits());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		try {
			// 卡住了
			t3.acquire();
			cs.accept("5 availablePermits=" + t3.availablePermits());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		Runtime.getRuntime().addShutdownHook(new Thread(()->{
			// 主线程阻塞时，关闭程序，此句不输出
			cs.accept("------test3() end");
		}));
	}
	
	/**
	 * 试验4：初始许可证为负数时，汇总线程等待多个线程执行完毕再执行
	 * 感觉和 join()有些像了...
	 * @author ben
	 * @date 2021-09-10 23:21:27 CST
	 */
	public static void test4() {
		cs.accept("test4() start");
		ExecutorService es = Executors.newFixedThreadPool(10);
		
		// 初始为负数许可证
		// -8:9个线程release后，信号量的许可证数变为1，此时可以获取
		final Semaphore sp = new Semaphore(-8);
		
		// 1、建立9个线程release，时间随机
		IntStream.range(0, 9).forEach(item->{
			es.execute(()->{
				String tname = Thread.currentThread().getName();
				// 15秒
				int rand = current().nextInt(15);
				cs.accept(tname + ": rand=" + rand + ", date=" + new Date());
				try {
					TimeUnit.SECONDS.sleep(rand);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				
				// 释放
				sp.release();
				cs.accept(tname + ": release, availablePermits=" + sp.availablePermits());
			});
		});

		// 等待线程
		es.execute(()->{
			try {
				String tname = Thread.currentThread().getName();
				cs.accept(tname + ": 等待start, date=" + new Date());
				sp.acquire();
				cs.accept(tname + ": 等到了！date=" + new Date());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});

		Runtime.getRuntime().addShutdownHook(new Thread(()->{
			cs.accept("------test4() end");
		}));
	}
}

/**
 * 试验1
 * @author ben
 * @date 2021-09-10 21:30:03 CST
 */
class SemaphoreTest1 {

	private static Consumer<Object> cs = System.out::println;

	public void basicUse() throws InterruptedException {
		// 仅1个许可证
		final Semaphore sp1 = new Semaphore(1);
		cs.accept("1、availablePermits=" + sp1.availablePermits() + ", hasQueuedThreads=" + sp1.hasQueuedThreads());
		
		// step.0
		sp1.acquire();
		cs.accept("2、availablePermits=" + sp1.availablePermits() + ", hasQueuedThreads=" + sp1.hasQueuedThreads());
		
		// step.1
		// 许可证为0，阻塞，3无法执行
//		sp1.acquire();
//		cs.accept("3、availablePermits=" + sp1.availablePermits() + ", hasQueuedThreads=" + sp1.hasQueuedThreads());
		
		// step.2
		boolean try1 = sp1.tryAcquire();
		cs.accept("4、try1=" + try1);
		cs.accept("4、availablePermits=" + sp1.availablePermits() + ", hasQueuedThreads=" + sp1.hasQueuedThreads());
		
		// step.3
		sp1.release();
		cs.accept("5、availablePermits=" + sp1.availablePermits() + ", hasQueuedThreads=" + sp1.hasQueuedThreads());
		
		// ---
		
		// 多个许可证
		final Semaphore sp2 = new Semaphore(10);
		cs.accept("6、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		sp2.acquire(5);
		cs.accept("7、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		sp2.acquire();
		cs.accept("8、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		// 阻塞：没有足够的许可证了，，如果此时另外的线程释放了许可证，则可能获取到
//		sp2.acquire(10);
//		cs.accept("9、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		sp2.acquire(sp2.availablePermits());
		cs.accept("10、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		// 多次释放
		sp2.release(3);
		cs.accept("11、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		sp2.release(3);
		cs.accept("12、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		// 意外：sp2总数10，连续3次释放了12个，结果，13的availablePermits=12！！！动态增加？TODO
		// 可能导致编程错误
		sp2.release(6);
		cs.accept("13、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		// 获取12个成功了！！
		sp2.acquire(12);
		cs.accept("14、availablePermits=" + sp2.availablePermits() + ", hasQueuedThreads=" + sp2.hasQueuedThreads());
		
		
		// ---
		
		// 意外：可以设置为负数！
		final Semaphore sp3 = new Semaphore(-1);
		cs.accept("15、availablePermits=" + sp3.availablePermits() + ", hasQueuedThreads=" + sp3.hasQueuedThreads());
		// 获取失败
		boolean try3 = sp3.tryAcquire();
		cs.accept("try3=" + try3);
		cs.accept("16、availablePermits=" + sp3.availablePermits() + ", hasQueuedThreads=" + sp3.hasQueuedThreads());
		
		// 许可证数量为负的信号量 释放
		sp3.release();
		// availablePermits=0
		cs.accept("17、availablePermits=" + sp3.availablePermits() + ", hasQueuedThreads=" + sp3.hasQueuedThreads());
		sp3.release(20);
		// availablePermits=20
		cs.accept("18、availablePermits=" + sp3.availablePermits() + ", hasQueuedThreads=" + sp3.hasQueuedThreads());
		
		// 获取成功
		sp3.acquire();
		cs.accept("19、availablePermits=" + sp3.availablePermits() + ", hasQueuedThreads=" + sp3.hasQueuedThreads());
		// 获取成功
		sp3.acquire(10);
		cs.accept("20、availablePermits=" + sp3.availablePermits() + ", hasQueuedThreads=" + sp3.hasQueuedThreads());
		
		// 总结，
		// 信号量的参数 permits，是初始值，可以使用 release 增大——多释放一下；
		// 许可证可以为负数（或0），此时无法获取到许可证，acquire阻塞；
		// 许可证为负数时，可以用在等待多个任务执行后，再执行下一个任务的场景；
		// 获取许可证后，不释放或者少释放许可证，可能导致总的许可证数量减少；
		
	}
	
}

/**
 * 试验2
 * @author ben
 * @date 2021-09-10 22:07:28 CST
 */
class SemaphoreTest2 {

	private static Consumer<Object> cs = System.out::println;

	// 共享资源：final, 不变
	private final String star = "❤";
	
	// 最多5个线程可以使用 star
	private final int initPermits = 5;
	// 非公平
//	private final Semaphore shareSp =  new Semaphore(initPermits);
	// 公平方式
	private final Semaphore shareSp =  new Semaphore(initPermits, true);
	
	private long useCnt = 0;
	private long failedCnt = 0;
	
	/**
	 * 使用star
	 * @author ben
	 * @throws InterruptedException 
	 * @date 2021-09-10 22:11:10 CST
	 */
	public void useStar() throws InterruptedException {
		String tname = Thread.currentThread().getName();
		// availablePermits=5时，可能会有排队线程
		cs.accept(tname + ": 1、permits=" + shareSp.availablePermits() + ", queuedLen=" + shareSp.getQueueLength());
		
		// 尝试获取：不会产生排队等候线程，，queuedLen一直为0
//		boolean try1 = shareSp.tryAcquire();
		// 会抛出 InterruptedException
		boolean try1 = shareSp.tryAcquire(500, TimeUnit.MILLISECONDS);
		cs.accept(tname + ": 2、permits=" + shareSp.availablePermits() + ", queuedLen=" + shareSp.getQueueLength());
		if (try1) {
			// 获取成功，使用start
			cs.accept(tname + ": 3 use " + star + ", haha");
			
			try {
				// 固定值时，会一起release许可证，改为随机值
				TimeUnit.SECONDS.sleep(current().nextInt(4));
			} catch (InterruptedException e) {
				cs.accept(tname + ": 4 异常=" + e.getMessage());
			}

			useCnt++;
			shareSp.release();
			cs.accept(tname + ": 5、permits=" + shareSp.availablePermits() + ", queuedLen=" + shareSp.getQueueLength());
		} else {
			failedCnt++;
			cs.accept(tname + ": 6 failed...");
		}
		
		// 可以看到 得到许可 和 获取失败的次数对比
		cs.accept(tname + ": 7 end...useCnt=" + useCnt + ", failedCnt=" + failedCnt);
	}
	
}

/**
 * 试验3：增强Semaphore
 * 既然是增强，就要重写Semaphore类的所有方法，这样，增强的SemaphoreTest3才是一个真正的Semaphore；
 * 否则，使用组合Semaphore对象的方式来实现即可
 * 第一版：暂实现 acquire、release版本
 * @author ben
 * @date 2021-09-10 23:01:44 CST
 */
class SemaphoreTest3 extends Semaphore {

	private static Consumer<Object> cs = System.out::println;
	
	// 两个构造函数：直接调用 super
	
	public SemaphoreTest3(int permits) {
		super(permits);
	}

	public SemaphoreTest3(int permits, boolean fair) {
		super(permits, fair);
	}
	
	// 存储获取了许可的线程
	private final ConcurrentLinkedQueue<Thread> queue = new ConcurrentLinkedQueue<Thread>();
	
	// 重写所有函数Semaphore公开函数
	
	@Override
	public void acquire() throws InterruptedException {
		String tname = Thread.currentThread().getName();
		cs.accept(tname + ": acquire 1-" + new Date());
		super.acquire();
		cs.accept(tname + ": acquire 2-" + new Date());
		// 将线程加入队列
		queue.add(Thread.currentThread());
		cs.accept(tname + ": acquire 3-" + new Date() + ", queue.size=" + queue.size());
	}
	
	/**
	 * 先判断是否已获取许可证，获取了才可以移除
	 */
	@Override
	public void release() {
		final Thread th = Thread.currentThread();
		String tname = Thread.currentThread().getName();
		cs.accept(tname + ": release 1-" + new Date());
		if (!queue.contains(th)) {
			cs.accept(tname + ": 无权释放");
			return;
		}
		
		super.release();
		cs.accept(tname + ": release 2-" + new Date());
		
		// 释放完成，将线程移除队列
		queue.remove(th);
		cs.accept(tname + ": release 3-" + new Date() + ", queue.size=" + queue.size());
	}
	
	// 重写未完待续 TODO
	
}
